package style.JUC多线程与并发.ProducerConsumerWithBlockingQueue;

import org.apache.commons.lang3.StringUtils;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 线程通信——阻塞队列版
 * 模拟实现生产者消费者
 * 阻塞队列
 */
class MyRescourse {
    // 判断  生产还是消费
    // true  表示生产
    // false 表示消费
    private volatile boolean FLAG = true;

    // 使用原子整形包装类 add or remove
    private AtomicInteger atomicInteger = new AtomicInteger();

    // 定义一个阻塞队列实现生产者往里面丢数据  消费者往里面拿数据
    BlockingQueue<String> blockingQueue = null;

    // 这里构造方法传入阻塞队列
    public MyRescourse(BlockingQueue<String> blockingQueue) {
        this.blockingQueue = blockingQueue;
    }

    // 实现生产者方法
    public void myProduce() throws InterruptedException {
        // 插入队列的数据
        String date = null;
        // 判断插入队列是否成功的标志
        boolean resFlag;

        while (FLAG) {
            date = atomicInteger.incrementAndGet() + "";
            resFlag = blockingQueue.offer(date, 2L, TimeUnit.SECONDS);
            if (resFlag) {
                System.out.println(Thread.currentThread().getName()+"插入队列成功！-->"+date);
            }else {
                System.out.println(Thread.currentThread().getName()+"插入队列失败！-->"+date);
            }
            TimeUnit.SECONDS.sleep(2);
        }
        System.out.println(Thread.currentThread().getName()+"生产者进来！");
    }

    // 实现消费者方法
    public void myConsumer() throws InterruptedException {
        String valRes;
        while (FLAG){
            valRes = blockingQueue.poll(3L, TimeUnit.SECONDS);
            if (StringUtils.isNotBlank(valRes)){
                System.out.println(Thread.currentThread().getName()+"获取队列数据成功！--"+valRes);
            }else {
                System.out.println(Thread.currentThread().getName()+"获取队列数据null！--");
                FLAG = false;
                return;
            }
        }
    }

    // stop 方法
    public void stop(){
        this.FLAG = false;
    }
}


public class ProducerConsumerWithBlockingQueueDemo {
    public static void main(String[] args) {
        MyRescourse myRescourse = new MyRescourse(new ArrayBlockingQueue<>(10));

        for (int i = 0; i < 2; i++) {
            new Thread(()->{
                try {
                    myRescourse.myProduce();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }, "produce"+String.valueOf(i)).start();
        }

        for (int i = 0; i < 2; i++) {
            new Thread(()->{
                try {
                    myRescourse.myConsumer();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }, "consumer"+String.valueOf(i)).start();
        }
    }
}
